#include "SchedulerInc.h"

using namespace bimar_scheduler_lib;

/*
 * CJobThreadPoolProcess
 */
gulong CJobThreadPoolProcess::RunProcess()
{
  g_return_val_if_fail(m_task, -1);
  
  /* sync do the task */
  m_task->Run();
  
  return 0;
}

/*
 * CJobScheduler
 */
CJobScheduler::CJobScheduler(gboolean taskOwner)
:CGThreadWrapper("CJobScheduler")
,m_threadPool(NULL)
,m_doneJobs(0)
{
  m_job = new CJob(taskOwner);
}

CJobScheduler::~CJobScheduler()
{
  Destroy();

  if(m_job)
    delete m_job;
  m_job = NULL;
}

gboolean CJobScheduler::Create()
{
  glong tasks = m_job->GetSize();
  return m_threadPool.CreatePool(tasks, TRUE /*exclussive*/);
}

gboolean CJobScheduler::Destroy()
{
  /* thread manager */
  m_threadPool.DestroyPool(TRUE /*immediate*/, TRUE /*wait*/);

  /* thread */
  DestroyThread();

  return TRUE;
}

gboolean CJobScheduler::PushBack(CTask *task)
{
  g_return_val_if_fail(task && m_job, FALSE);

  m_job->PushBack(task);

  return TRUE;
}

gboolean CJobScheduler::PushBack(CJob *job)
{
  g_return_val_if_fail(job && m_job, FALSE);

  CGAutoLock lock(&(job->m_taskListCS));

  for(TaskListIt it=job->m_taskList.begin(); it!=job->m_taskList.end(); it++)
  {
    CTask *task = *it;
	PushBack(task);
  }

  return TRUE;
}

gboolean CJobScheduler::Start()
{
  BindTasks();

  return CreateThread();
}

gboolean CJobScheduler::Stop()
{
  CGAutoLock lock(&(m_job->m_taskListCS));

  for(TaskListIt it=m_job->m_taskList.begin(); it!=m_job->m_taskList.end(); it++)
  {
	CTask *task = *it;
	if(task &&
	 ((task->GetStatus() == EStatus_Running) || (task->GetStatus() == EStatus_Paused)) )
	  task->Stop();
  }

  return TRUE;
}

gboolean CJobScheduler::Pause()
{
  CGAutoLock lock(&(m_job->m_taskListCS));

  for(TaskListIt it=m_job->m_taskList.begin(); it!=m_job->m_taskList.end(); it++)
  {
	CTask *task = *it;
	if(task &&
	 ((task->GetStatus() == EStatus_Running) || (task->GetStatus() == EStatus_Stopped)) )
	  task->Pause();
  }

  return TRUE;
}

gboolean CJobScheduler::Resume()
{
  CGAutoLock lock(&(m_job->m_taskListCS));

  for(TaskListIt it=m_job->m_taskList.begin(); it!=m_job->m_taskList.end(); it++)
  {
	CTask *task = *it;
	if(task && 
	 ((task->GetStatus() == EStatus_Paused) || (task->GetStatus() == EStatus_Stopped)) )
	  task->Resume();
  }

  return TRUE;
}

gboolean CJobScheduler::Abort()
{
  CGAutoLock lock(&(m_job->m_taskListCS));

  if(m_job->m_taskList.size()>0)
  {
    TaskListIt it = m_job->m_taskList.end();
    do
    {
      it--;

	  CTask *task = *it;
	  if(task)
	  {
	    /* not in threadpool. Force done for thread pool */
        if(task->GetStatus() == EStatus_Ready)
	    {
	      task->Abort();
	      m_doneJobs++;
	    }
	    else if( (task->GetStatus() == EStatus_Ready) || (task->GetStatus() == EStatus_Running) || (task->GetStatus() == EStatus_Paused) || (task->GetStatus() == EStatus_Stopped) )
	      task->Abort();
      }
    
	  if(it == m_job->m_taskList.begin())
	    break;
    }
    while(1);
  }
  
  return TRUE;
}

gboolean CJobScheduler::BindTasks()
{
  g_return_val_if_fail(m_job, FALSE);

  CGAutoLock lock(&(m_job->m_taskListCS));

  for(TaskListIt it=m_job->m_taskList.begin(); it!=m_job->m_taskList.end(); it++)
  {
	CTask *task = *it;
    TaskListIt nextIt = it; nextIt++;
	if(nextIt != m_job->m_taskList.end())
	{
	  CTask *nextTask = *nextIt;
	  if(task && nextTask && task->IsParallelNext())
        nextTask->SetPropertyPrevParallel(TRUE);
	}

	/* status ready for all tasks */
	if(task)
	  task->Ready();
  }

  return TRUE;
}

gulong CJobScheduler::RunThread()
{
  g_return_val_if_fail(m_job, -1);

  glong totalTasks = m_job->GetSize();
  CTask *prevTask = NULL;
  gboolean taskSubmit = TRUE;
  glong i = 0;
  TaskList submitList;
  while(i<totalTasks)
  {
    CTask *task = m_job->Get(i);
    taskSubmit = FALSE;

	/* check task is ready */
	if(task->GetStatus() != EStatus_Ready)
	{
	  task = NULL;
	  i++;
	}

    /* if prev==NULL or current paralle with previous */
	if(task && (!prevTask || task->IsParallelPrev()))
    {
	  CJobThreadPoolProcess *process = new CJobThreadPoolProcess(task);
	  taskSubmit = m_threadPool.Submit(process, FALSE);
	  if(!taskSubmit)
	    delete process;
	  else
	    submitList.push_back(task);
      i++;
      prevTask = task;
    }
    else
      prevTask = NULL;

	/* execute tasks */
	if( (!taskSubmit && m_threadPool.GetNumSubmitThreads()>0)
     ||	(taskSubmit && (i>=totalTasks)) )
    {
      m_threadPool.Push();
      m_threadPool.WaitRunningDone();

	  /* to DONE if no aborted */
	  for(TaskListIt it = submitList.begin(); it!=submitList.end(); it++)
	  {
	    CTask *task = *it;
		if(task->GetStatus() == EStatus_Running)
		  task->Done();
	    m_doneJobs++;
	  }
	  submitList.clear();
	}
  } 

  return 0;
}

gboolean CJobScheduler::WaitJobDone(gulong milis)
{
  g_return_val_if_fail(m_job, FALSE);

  glong jobs = m_job->GetSize();

  gboolean timeout = FALSE;
  GTimer *timer = NULL;
  if(milis != WAIT_INFINITE)
  {
	timer = g_timer_new();
	g_timer_start(timer);
  }
  
  while(m_doneJobs < jobs)
  {
	if(milis != WAIT_INFINITE)
      timeout = (g_timer_elapsed(timer, NULL) * 1000 >= milis);
    
	g_thread_yield();
  }

  return !timeout;
}
